package cn.itcast.flink.source;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;


import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.concurrent.TimeUnit;

/**
 * @author lilulu
 */
public class StreamSourceMysqlDemo {

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }

    /**
     * 自定义数据源，从MySQL表中加载数据，并且实时增量加载，每隔5秒钟加载一次
     */
    public static class MysqlSource implements ParallelSourceFunction<Student> {
        private boolean isRunning = true;


        public void run(SourceContext<Student> sourceContext) throws Exception {
            while (isRunning) {
                /**
                 * 1、注册驱动
                 * 2、获取连接
                 * 3、获取预编译对象
                 * 4、执行sql
                 * 5、处理结果集
                 * 6、关闭资源
                 */
                Class.forName("com.mysql.jdbc.Driver");
                Connection connection = DriverManager.getConnection("jdbc:mysql://node1.itcast.cn:3306/?useSSL=false", "root", "123456");
                PreparedStatement preparedStatement = connection.prepareStatement("select id,name,age from db_flink.t_student");
                ResultSet resultSet = preparedStatement.executeQuery();
                while (resultSet.next()) {
                    int id = resultSet.getInt("id");
                    String name = resultSet.getString("name");
                    int age = resultSet.getInt("age");
                    Student student = new Student(id, name, age);
                    sourceContext.collect(student);
                }
                resultSet.close();
                preparedStatement.close();
                connection.close();

                //每隔5秒加载一次数据
                TimeUnit.SECONDS.sleep(5);
            }

        }

        public void cancel() {
            isRunning = false;
        }
    }

    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 数据源-source
        DataStreamSource<Student> studentDataStreamSource = env.addSource(new MysqlSource());
        // 3. 数据转换-transformation
        // 4. 数据终端-sink
        studentDataStreamSource.printToErr();
        // 5. 触发执行-execute
        env.execute("StreamSourceMysqlDemo");
    }
}